package com.tarena.test.rabbitmq.topic;

import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.Delivery;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
import org.junit.Before;
import org.junit.Test;

//测试主题模式的类
public class TopicMode {
    //交换机名称常量
    public static final String TOPIC_EX="topic-exchange";
    //交换机类型常量
    public static final String TOPIC_TYPE="topic";
    //队列名称常量
    public static final String TOPIC_QUEUEA="topic-queueA";
    public static final String TOPIC_QUEUEB="topic-queueB";
    //路由键常量
    public static final String TOPIC_ROUTINGKEYA="中国.#";
    public static final String TOPIC_ROUTINGKEYB="*.上海.*";
    private Channel channel;
    @Before
    public void initChannel() throws IOException, TimeoutException {
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        factory.setVirtualHost("/");
        Connection connection= factory.newConnection();
        channel=connection.createChannel();
    }
    //声明交换机 绑定队列
    @Test
    public void declareComponent() throws IOException {
        channel.exchangeDeclare(TOPIC_EX, TOPIC_TYPE);
        channel.queueDeclare(TOPIC_QUEUEA, false, false, false, null);
        channel.queueDeclare(TOPIC_QUEUEB, false, false, false, null);
        channel.queueBind(TOPIC_QUEUEA, TOPIC_EX, TOPIC_ROUTINGKEYA);
        channel.queueBind(TOPIC_QUEUEB, TOPIC_EX, TOPIC_ROUTINGKEYB);
    }
    //绑定queuea的消费者 简化代码 直接使用lamda
    @Test
    public void consumerA() throws IOException {
        channel.basicConsume(TOPIC_QUEUEA, true, (consumerTag, message) -> {
            //消费逻辑 解析message中的消息体 打印
            String msg=new String(message.getBody());
            System.out.println("消费者A拿到消息:"+msg);
            //获取消息的路由key
            String key = message.getEnvelope().getRoutingKey();
            System.out.println("消费者A的消息路由key:"+key);
        }, consumerTag -> {
            //消费者消失的 代码 TODO
        });
        while(true);
    }
    @Test
    public void consumerB() throws IOException {
        channel.basicConsume(TOPIC_QUEUEB, true, (consumerTag, message) -> {
            //消费逻辑 解析message中的消息体 打印
            String msg=new String(message.getBody());
            System.out.println("消费者B拿到消息:"+msg);
            //获取消息的路由key
            String key = message.getEnvelope().getRoutingKey();
            System.out.println("消费者B的消息路由key:"+key);
        }, consumerTag -> {
            //消费者消失的 代码 TODO
        });
        while (true);
    }
    @Test
    public void sendMessage() throws IOException {
        System.out.println("生产者即将发送");
        //1.准备一个发送的消息数据
        String msg="这是我的第一条消息";
        channel.basicPublish(TOPIC_EX,"中国.北京.朝阳.望京.葫芦村",null,msg.getBytes());
        System.out.println("生产者发送结束");
    }
}
